#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/8/29 16:50
# @Author  : youfeng
import json
import threading
import time
from Queue import Queue

from beanstalkc import SocketError, CommandFailed

from mq.pybeanstalk import PyBeanstalk


class BeanstalkHandler(threading.Thread):
    # 阻塞时间
    __PAUSE_BLOCK_TIME = 10

    # 休眠时间
    __PAUSE_NORMAL_TIME = 0.5

    # 最大待发送队列大小
    __MAX_QUEUE_SIZE = 50

    # 最大容许消息队列阻塞数目
    __MAX_BEANSTALK_BLOCK_SIZE = 1000

    def __init__(self, host, port, log):
        super(BeanstalkHandler, self).__init__()
        if log is None:
            raise Exception("缺少log模块...")

        self.log = log

        if not isinstance(host, basestring):
            raise Exception("消息队列连接参数异常: host = {}".format(host))
        if not isinstance(port, int):
            raise Exception("消息队列连接参数异常: port = {}".format(port))
        self.__beanstalk = PyBeanstalk(host, port)

        # 输送队列
        self.__queue = Queue()

        # 消息队列是否正常运行
        self.__is_running = True

        # 启动消息队列线程
        self.setDaemon(True)
        self.start()

    def __del__(self):
        self.close()

    def close(self):
        self.__is_running = False
        self.__queue.put_nowait('beanstalk close')

    # 判断内存队列是否有阻塞
    def __wait_mem_queue(self):

        count = 0
        while True:
            count += 1
            q_size = self.__queue.qsize()
            if q_size >= self.__MAX_QUEUE_SIZE:
                if count % 100 == 0:
                    self.log.info("当前线程队列阻塞, 需要休眠: queue size = {}".format(q_size))

                time.sleep(0.2)
                continue
            break

    def push_msg(self, tube, data_item):

        if not self.__is_running:
            return False

        # 判断是否需要等待
        self.__wait_mem_queue()

        msg = json.dumps(data_item)
        self.__queue.put_nowait((tube, msg))
        return True

        # 判断是否需要暂停

    def __wait_beanstalk_queue(self, tube):
        times = 0
        while True:
            sleep_time = self.__PAUSE_NORMAL_TIME
            try:
                count = self.__beanstalk.get_tube_count(tube)
            except CommandFailed, (_, status, results):
                if status == 'NOT_FOUND':
                    self.log.error('获取消息队列不存在: tube = {}'.format(tube))
                    count = 0
                else:
                    self.log.error('获取消息队列异常: tube = {} status = {}'.format(tube, status))
                    sleep_time = self.__PAUSE_BLOCK_TIME
                    count = self.__MAX_BEANSTALK_BLOCK_SIZE
            except Exception as e:
                sleep_time = self.__PAUSE_BLOCK_TIME
                count = self.__MAX_BEANSTALK_BLOCK_SIZE
                self.log.error("获取消息队列大小失败: ")
                self.log.exception(e)

            if count < self.__MAX_BEANSTALK_BLOCK_SIZE:
                return True

            times += 1
            if times % 20 == 0:
                self.log.info("当前消息队列阻塞需要休眠: tube = {} sleep = {}".format(tube, sleep_time))

            # 开始休眠
            time.sleep(sleep_time)

    def run(self):
        self.log.info('开始运行消息队列...')
        while True:
            try:

                body = self.__queue.get()

                # 如果是字符串则直接退出，视为退出消息
                if isinstance(body, basestring):
                    self.log.info("消息队列线程收到字符串数据, 退出线程: body = {}".format(body))
                    break

                if not isinstance(body, tuple):
                    self.log.error("消息类型错误: body type = {} msg = {}".format(type(body), body))
                    time.sleep(5)
                    continue

                tube, msg = body
                while True:
                    try:
                        # 发送前先判断是否需要休眠 如果阻塞则直接返回
                        if not self.__wait_beanstalk_queue(tube):
                            break

                        result = self.__beanstalk.put(tube, msg)
                        if result is None:
                            self.log.error("数据量太大: msg len = {}".format(len(msg)))

                        break
                    except SocketError as e:
                        time.sleep(10)
                        self.__beanstalk.reconnect()
                        self.log.warn("reconnect beanstalk...")
                        self.log.exception(e)
                    except Exception as e:
                        self.log.error('捕获异常休眠:')
                        self.log.exception(e)
                        time.sleep(10)
            except Exception as e:
                self.log.info('当前队列大小: size = {}'.format(self.__queue.qsize()))
                self.log.exception(e)
                time.sleep(5)
        self.log.info("消息队列正常退出...")
